package com.czk.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 带回调的发送消息
 *
 * @Author:ChenZhangKun
 * @Date: 2021/5/23 14:54
 */
public class CallBackProducer {
    public static void main(String[] args) {
        // 创建配置信息
        Properties properties = new Properties();
        // kafka服务器
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.25.133:9092");
        // 序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // VALUE序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 发送数据
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("aaa", "aaa++" + i), (metadata, exception) -> {
                if (exception == null) {
                    // 发送成功
                    // 打印主题
                    System.out.println("topic=" + metadata.topic());
                    // 打印偏移量
                    // 偏移量从0开始
                    long offset = metadata.offset();
                    System.out.println("offset=" + offset);
                    // 打印分区
                    int partition = metadata.partition();
                    System.out.println("partition=" + partition);
                } else {
                    exception.printStackTrace();
                }
            });
        }
        // 关闭资源
        kafkaProducer.close();
    }
}
